#!/usr/bin/env python
# -*- coding:utf-8 -*-

import threading,logging


# Barrier实例
"""
# 输出格式定义
FORMAT = '%(asctime)-15s\t [%(threadName)s,%(thread)8d] %(message)s'
logging.basicConfig(level=logging.INFO,format=FORMAT)

def worker(barrier:threading.Barrier):
    logging.info('waiting for {} threads.'.format(barrier.n_waiting))
    try:
        barrier_id = barrier.wait()
        logging.info('after barrier {}'.format(barrier_id))
    except threading.BrokenBarrierError:
        logging.info('Broken Barrier')


barrier = threading.Barrier(3)

for x in range(3): 
    threading.Thread(target=worker,name='worker-{}'.format(x),args=(barrier,)).start()

logging.info('started')
'''
从运行结果看出:
所有线程冲到了Barrier前等待,直到到达parties的数目,屏障打开,所有线程停止等待,继续执行.
再有线程wait,屏障就绪等到到达参数方数目.
'''
"""


# 举例: 赛马比赛所有马匹就位,开闸. 下一批马匹陆续来到闸门前等待比赛.
"""
# 输出格式定义
FORMAT = '%(asctime)-15s\t [%(threadName)s,%(thread)8d] %(message)s'
logging.basicConfig(level=logging.INFO,format=FORMAT)

def worker(barrier:threading.Barrier):
    logging.info('waiting for {} threads.'.format(barrier.n_waiting))
    try:
        barrier_id = barrier.wait()
        logging.info('after barrier {}'.format(barrier_id))
    except threading.BrokenBarrierError:
        logging.info('Broken Barrier. run.')


barrier = threading.Barrier(3)

for x in range(0,9):
    if x == 2:
        barrier.abort()
    elif x == 6:
        barrier.reset()
    threading.Event().wait(1)
    threading.Thread(target=worker,name='worker-{}'.format(x),args=(barrier,)).start()
"""
"""
上例中,屏障中等待了2个,屏障就被break了,waiting的线程抛了BrokenBarrierError异常,新wait的线程也抛异常,直到屏障恢复，
才继续按照parties数目要求继续拦截线程.
"""




# wait方法超时实例
# 如果wait方法超时发生,屏障将处于broken状态,直到reset。
"""
# 输出格式定义
FORMAT = '%(asctime)-15s\t [%(threadName)s, %(thread)8d] %(message)s'
logging.basicConfig(level=logging.INFO,format=FORMAT)

def worker(barrier:threading.Barrier,i:int):
    logging.info('waiting for {} threads.'.format(barrier.n_waiting))
    try:
        logging.info(barrier.broken) # 是否broken
        if i<3:
            barrier_id = barrier.wait(1) # 超时后,屏障broken
        else:
            if i == 6:
                barrier.reset() # 恢复屏障
            barrier_id = barrier.wait()
        logging.info('after barrier {}'.format(barrier_id))
    except threading.BrokenBarrierError:
        logging.info('Broken Barrier. run.')

barrier = threading.Barrier(3)

for x in range(0,9):
    threading.Event().wait(2)
    threading.Thread(target=worker,name='worker-{}'.format(x),args=(barrier,x)).start()
"""



# semaphore信号量
"""
import threading,logging,time
# 输出格式定义
FORMAT = '%(asctime)-15s\t [%(threadName)s,%(thread)8d] %(message)s'
logging.basicConfig(level=logging.INFO,format=FORMAT)

def worker(s:threading.Semaphore):
    logging.info('in sub thread')
    logging.info(s.acquire()) # 阻塞
    logging.info('sub thread over')

# 信号量
s = threading.Semaphore(3)
logging.info(s.acquire())
print(s._value)
logging.info(s.acquire())
print(s._value)
logging.info(s.acquire())
print(s._value)

threading.Thread(target=worker,args=(s,)).start()
time.sleep(2)

logging.info(s.acquire(False))
logging.info(s.acquire(timeout=3)) # 阻塞3秒

# 释放
logging.info('released')
s.release()
"""


# 应用举例
"""
连接池: 因为资源有限,且开启一个连接成本高,所以,使用连接池.

一个简单的连接池: 连接池应该有容量(总数),
有一个工厂方法可以获取连接,能够把不用的连接返回,供其他调用者使用.
"""
"""
class Conn:
    def __init__(self,name):
        self.name=name

class Pool:
    def __init__(self,count:int):
        self.count = count
        # 池中是连接对象的列表
        self.pool = [self._connect("conn-{}".format(x)) for x in range(self.count)]

    def _connect(self,conn_name):
        # 创建连接的方法,返回一个名称
        return Conn(conn_name)
    
    def get_conn(self):
        # 从池中拿走一个连接
        if len(self.pool) > 0:
            return self.pool.pop()

    def return_conn(self,conn:Conn):
        # 向池中添加一个连接
        self.pool.append(conn)
"""
"""
真正的连接池的实现比上面的例子要复杂的多,这里只是简单的一个功能的实现.
本例中,get_conn()方法在多线程的时候有线程安全问题.
假设池中正好有一个连接,有可能多个线程判断池的长度是大于0的,当一个线程拿走了连接对象,
其他线程再来pop就会抛异常的。 如何解决?
1. 加锁,在读写的地方加锁.
2. 使用信号量Semaphore
"""



# 使用信号量对上例进行修改
"""
import threading,logging,random

FORMAT = "%(asctime)s %(thread)d %(threadName)s %(message)s"
logging.basicConfig(level=logging.INFO, format=FORMAT)

class Conn:
    def __init__(self,name):
        self.name = name

    def __repr__(self):
        return self.name
    
class Pool:
    def __init__(self,count:int):
        self.count = count
        # 池中是连接对象的列表
        self.pool = [self._connect("conn-{}".format(x)) for x in range(count)]
        self.semaphore = threading.Semaphore(count) # threading.Semaphore()

    def _connect(self,conn_name):
        # 返回一个名称
        return Conn(conn_name)

    def get_conn(self):
        # 从池中拿走一个连接
        print('------------------')
        self.semaphore.acquire()
        print('='*10)
        conn = self.pool.pop()
        return conn

    def return_conn(self,conn:Conn):
        # 向池中添加一个连接
        self.pool.append(conn)
        self.semaphore.release()

# 连接池初始化
pool = Pool(3)

def worker(pool:Pool):
    conn = pool.get_conn()
    logging.info(conn)
    # 模拟使用了一段时间
    threading.Event().wait(random.randint(1,4))
    pool.return_conn(conn)

for i in range(6):
    threading.Thread(target=worker,name="worker-{}".format(i),args=(pool,)).start()
"""
"""
上例中,使用信号量解决资源有限的问题
如果池中有资源,请求者获取资源时信号量减1,拿走资源. 当请求超过资源数,请求者只能等待.
当使用者用完归还资源后信号量加1,等待线程就可以被唤醒拿走资源.
# 注意: 这个例子不能用到生产环境,只是为了说明信号量使用的例子,还有很多未完成功能.
"""


# 问题: self.conns.append(conn) 这一句要不要加锁
# 1. 从程序逻辑上分析
# 1.1 假设如果还没有使用信号量,就release,会怎么样?
"""
sema = threading.Semaphore(3)
logging.warning(sema.__dict__)
for i in range(3):
    sema.acquire()
logging.warning('~'*15)
logging.warning(sema.__dict__)

for i in range(4):
    sema.release()
logging.warning(sema.__dict__)

for i in range(3):
    sema.acquire()
logging.warning('~'*20)
logging.warning(sema.__dict__)
sema.acquire()
logging.warning('~'*15)
logging.warning(sema.__dict__)
"""
# 从上例输出结果可以看出,竟然内置计数器达到了4,这样实际上超出我们的最大值,需要解决这个问题.






# 数据结构和GIL
"""
import queue

q = queue.Queue(8)

if q.qsize() == 7:
    q.put() # 上下两句可能被打断.
if q.qsize() == 1:
    q.get() #未必会成功.
"""





# GIL全局解释器锁
import logging,datetime

logging.basicConfig(level=logging.INFO,format="%(thread)s %(message)s")
start = datetime.datetime.now()

# 计算
def calc():
    sum = 0
    for _ in range(1000000000):
        sum += 1

calc()
calc()
calc()
calc()
calc()

delta = (datetime.datetime.now() - start).total_seconds()
logging.info(delta)



import threading,logging,datetime

logging.basicConfig(level=logging.INFO,format="%(thread)s %(message)s")
start = datetime.datetime.now()

# 计算
def calc():
    sum = 0
    for _ in range(1000000000):
        sum += 1

t1 = threading.Thread(target=calc)
t2 = threading.Thread(target=calc)
t3 = threading.Thread(target=calc)
t4 = threading.Thread(target=calc)
t5 = threading.Thread(target=calc)
t1.start()
t2.start()
t3.start()
t4.start()
t5.start()
t1.join()
t2.join()
t3.join()
t4.join()
t5.join()

delta = (datetime.datetime.now() - start).total_seconds()
logging.info(delta)
# 从两段程序测试结果看，Cpython中多线程没有优势,和一个线程执行时间相当.因为GIL的存在.
